package cn.hy.dataSync.task.runner;



import cn.hy.dataSync.config.DataBaseType;
import cn.hy.dataSync.config.ReaderConfig;
import cn.hy.dataSync.config.RecordSender;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.transport.record.ObjectColumn;
import cn.hy.dataSync.utils.RdbUtils;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;


@Slf4j
public class ReaderRunner extends AbstractRunner implements Runnable {

    private RecordSender recordSender;

    public ReaderRunner(TaskConfig taskConfig) {
        super(taskConfig);
    }

    public void setRecordSender(RecordSender recordSender) {
        this.recordSender = recordSender;
    }

    @Override
    public void run() {
        assert null != this.recordSender;
        try {
            log.debug("task reader starts to do init ...");
            init(); // 空实现

            log.debug("task reader starts to do prepare ...");
            prepare(); // 空实现

            log.debug("task reader starts to read ...");
            startRead(recordSender);

            recordSender.terminate();

            log.debug("task reader starts to do post ...");
            post();  // 空实现
        } catch (Throwable e) {
            log.error("Reader runner Received Exceptions:", e);
            super.markFail(e);
        }
    }

    @Override
    public void shutdown() {
        recordSender.shutdown();
    }

    private void init() {

    }

    private void prepare() {

    }

    private void post() {

    }

    private void startRead(RecordSender recordSender) {
        String querySql = this.getTaskConfig().getReaderSql();
        log.info("Begin to read record by Sql: [{}\n].", querySql);

        ReaderConfig readerConfig = this.getTaskConfig().getJobConfig().getReaderConfig();

        Connection conn = RdbUtils.getConnection(DataBaseType.MySql, readerConfig.getJdbcUrl(),
                readerConfig.getUsername(), readerConfig.getPassword());

        ResultSet rs = null;
        try {
            rs = RdbUtils.query(conn, querySql);

            ResultSetMetaData metaData = rs.getMetaData();
            int columnNumber = metaData.getColumnCount();

            // 读取的数据全部发送出去 chanel 中
            while (rs.next()) {
                transportOneRecord(recordSender, rs, metaData, columnNumber);
            }
            log.info("Finished read record by Sql: [{}\n]", querySql);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            RdbUtils.close(rs);
            RdbUtils.close(conn);
        }
    }

    private void transportOneRecord(RecordSender recordSender,
                                    ResultSet rs,
                                    ResultSetMetaData metaData,
                                    int columnNumber) {

        try {
            // 组装成 record 模型
            Record record = recordSender.createRecord();
            for (int i = 1; i <= columnNumber; i++) {
                record.addColumn(metaData.getColumnName(i), new ObjectColumn(rs.getObject(i)));
            }
            recordSender.sendToWriter(record);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
